package com.gxljc.bear.crawler.weixin;

import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import com.gxljc.commons.util.AdvFile;
import com.gxljc.commons.util.ILineParser;
import com.gxljc.commons.util.Pair;
import com.gxljc.commons.util.StringUtil;
import com.gxljc.bear.crawler.base.DolphinCrawlerConf;
import com.gxljc.bear.crawler.base.DolphinCrawlerConsts;
import com.gxljc.bear.crawler.base.DolphinFetchData;
import com.gxljc.bear.crawler.image.ImageFresh;
import com.gxljc.bear.crawler.proxy.ProxyUtil;
import com.gxljc.bear.crawler.util.SparkUtil;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.log4j.Logger;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;

import java.io.Serializable;
import java.util.*;

/**
 * 新榜微信账号数据爬虫。
 *
 * @author tanghaitao
 * @since 2022-10-9
 */
public class NewRankAccountCrawler implements Serializable {
    public static Logger LOG = Logger.getLogger(NewRankAccountCrawler.class);
    private Boolean proxy = false;
    private NewRankPageSeedCrawler crawler = null;
    private Boolean updateAll = false;

    public NewRankAccountCrawler(Boolean proxy, Boolean updateAll) {
        this.proxy = proxy;
        crawler = new NewRankPageSeedCrawler(proxy, updateAll);
    }

    public void crawl() throws Exception {
        final long SLEEP_TIME = 1 * 60 * 60 * 1000l;
        while (true) {
            List<AccountTable> seeds = null;
            try {
                seeds = WeixinUtil.getAccountSeed();
            } catch (Exception er) {
                er.printStackTrace();
            }
            if (CollectionUtils.isEmpty(seeds)) {
                LOG.error("page is empty, sleep " + SLEEP_TIME);
                Thread.sleep(SLEEP_TIME); //没有数据，暂停
                continue;
            }
            ProxyUtil.initQueue(proxy);
            int size = seeds.size();
            LOG.info("weixin account seeds size = " + size);
            try {
                crawl(seeds);
            } catch (Exception er) {
                er.printStackTrace();
                LOG.error("spark job error", er);
            }
            LOG.info("sleep " + SLEEP_TIME);
            Thread.sleep(SLEEP_TIME);
        }
    }

    public void crawlNoSpark(List<AccountTable> seeds) throws Exception {
        if (CollectionUtils.isEmpty(seeds)) return;
        for (AccountTable seed : seeds) {
            try {
                crawl(seed);
                Thread.sleep(2000);
            } catch (Exception er) {
                er.printStackTrace();
            }
        }
    }

    public void crawl(List<AccountTable> seeds) throws Exception {
        int coreMax = 30;
        JavaSparkContext jsc = SparkUtil.createCommonsSparkContext(
                "bear-新榜微信账号种子-" + seeds.size(), coreMax,
                coreMax * 2, SogouAccountCrawler.class);
        JavaRDD<AccountTable> seedsRDD = jsc.parallelize(new ArrayList<AccountTable>(seeds));
        long count = seedsRDD.mapPartitions(
                new FlatMapFunction<Iterator<AccountTable>, Integer>() {
                    private static final long serialVersionUID = 1L;

                    @Override
                    public Iterable<Integer> call(Iterator<AccountTable> seeds)
                            throws Exception {
                        int successCnt = 0;
                        while (seeds.hasNext()) {
                            try {
                                AccountTable seed = seeds.next();
                                int ret = crawl(seed);
                                if (ret == 1)
                                    successCnt++;
                                if (successCnt % 100 == 0) {
                                    LOG.info("successCnt=" + successCnt);
                                }
                            } catch (Exception er) {
                                er.printStackTrace();
                            }
                        }
                        return Arrays.asList(successCnt);
                    }
                }).count();
        LOG.info("all count=" + count);
        jsc.stop();
    }

    //单条url入口
    public int crawl(AccountTable table) throws Exception {
        LOG.info("id = " + table.accountId);
        String url = String.format(WeixinConst.NEWRANK_ACCOUNT_URL, table.accountId);
        DolphinFetchData fetchData = DolphinFetchData.getInstance(DolphinCrawlerConsts.CrawlerChannelType.MOVIE.getName());
        String html = fetchData.getEncodeHtml(url, proxy, WeixinConst.CRAWL_TIMEOUT);
        if (StringUtils.isEmpty(html)) {
            LOG.error("page result is empty url = " + url);
            return DolphinCrawlerConsts.RESULT_NO;
        }
        int flag = extract(table, html);
        if (flag == DolphinCrawlerConsts.RESULT_NO) {
            return DolphinCrawlerConsts.RESULT_NO;
        }
        if (StringUtils.isEmpty(table.uuid)) {
            LOG.error("account uuid is empty = " + url);
            return DolphinCrawlerConsts.RESULT_NO;
        }
        WeixinMongodbUtil.saveMongodbAccount(table, WeixinConst.MONGODB_TABLE_ACCOUNT);
        crawler.crawl(table);
        return DolphinCrawlerConsts.RESULT_YES;
    }

    private int extract(AccountTable table, String html) throws Exception {
        final Gson GSON = new Gson();
        String startStr = "var fgkcdg =";
        String end = "};";
        if (html.indexOf(startStr) < 0 || html.indexOf(end) < 0) {
            return DolphinCrawlerConsts.RESULT_NO;
        }
        int first = html.indexOf(startStr) + startStr.length();
        String json = html.substring(first, html.indexOf(end, first) + 1);
//        System.out.println("json = " + json);
        Map map = GSON.fromJson(json, new TypeToken<Map>() {
        }.getType());
        if (map == null || map.isEmpty()) return DolphinCrawlerConsts.RESULT_NO;
        return extractAccount(table, map);
    }


    //主体
    private void extractMainBody(AccountTable table, Map map) {
        Object certifiedObj = map.get("certified_text");
        if (certifiedObj == null) return;
        String certified = (String) certifiedObj;
        if (StringUtils.isEmpty(certified)) return;
        final String nouse = "微信认证：";
        if (certified.indexOf(nouse) >= 0) certified = certified.replace(nouse, "");
        table.setMainBody(certified);
    }

    //功能介绍
    private void extractDesc(AccountTable table, Map map) {
        Object descriptionObj = map.get("description");
        if (descriptionObj == null) return;
        String description = (String) descriptionObj;
        if (StringUtils.isEmpty(description)) return;
        table.setDesc(description);
    }

    //uuid
    private void extractUUID(AccountTable table, Map map) {
        Object uuidObj = map.get("uuid");
        if (uuidObj == null) return;
        String uuid = (String) uuidObj;
        if (StringUtils.isEmpty(uuid)) return;
        table.setUuid(uuid);
    }

    //抽取账号名称
    private void extractAccountName(AccountTable table, Map map) {
        Object nameObj = map.get("name");
        if (nameObj == null) return;
        String name = (String) nameObj;
        if (StringUtils.isEmpty(name)) return;
        table.setName(name);
    }

    //页面抽取并返回  object
    private int extractAccount(AccountTable table, Map map) throws Exception {
        extractAccountName(table, map);
        if (StringUtils.isEmpty(table.name)) return DolphinCrawlerConsts.RESULT_NO;
        extractMainBody(table, map);
        extractDesc(table, map);
        extractUUID(table, map);
        parseQRcode(table);
        return DolphinCrawlerConsts.RESULT_YES;
    }

    //生成二维码
    private void parseQRcode(AccountTable table) {
        String url = String.format(WeixinConst.WEIXIN_QR_URL, table.accountId);
        table.setQrCode(url);
        ImageFresh fresh = ImageFresh.getInstance();
        Pair<Integer, String> newPicPair = null;
        try {
            newPicPair = fresh.freshOne(url);
            String newPic = newPicPair.second;
            if (!url.equals(newPic))
                table.setQrCode(newPic);
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    public List<AccountTable> getSeed() throws Exception {
        final String filename = "weixin_all.txt";
        Configuration conf = DolphinCrawlerConf.getInstance();
        final List<AccountTable> seeds = new LinkedList<AccountTable>();
        AdvFile.loadFileInDelimitLine(
                conf.getConfResourceAsInputStream(filename), new ILineParser() {
                    @Override
                    public void parseLine(String line) {
                        String[] split = line.split(StringUtil.STR_DELIMIT_3RD);
                        if (split.length != 4) {
                            return;
                        }
                        String word1 = split[0].trim();
                        String word2 = split[1].trim();
                        String word3 = split[2].trim();
                        String word4 = split[3].trim();
                        if (word1.isEmpty() || word2.isEmpty() || word3.isEmpty() || word4.isEmpty()) {
                            return;
                        }
                        AccountTable table = new AccountTable();
                        table.setName(word1);
                        table.setAccountId(word2);
                        table.setAccountType(word3);
                        table.setTargetType(word4);
                        seeds.add(table);
                    }
                });
        return seeds;
    }
}
